package com.iflytek.controller;

import com.iflytek.bean.MonitorInfo;
import com.iflytek.schema.JSONDeserializationSchema;
import com.iflytek.utils.JdbcUtils;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Properties;


public class _05IllegalVehiclesTrackV1 {

    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);  //设置并行度

        //2.设置数据源
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers","hadoop102:9092");
        properties.setProperty("group.id","g4");


        FlinkKafkaConsumer<MonitorInfo> consumer = new FlinkKafkaConsumer<MonitorInfo>("topic-car",   new JSONDeserializationSchema<>(MonitorInfo.class),properties);
        DataStreamSource<MonitorInfo> ds1 = env.addSource(consumer);

        SingleOutputStreamOperator<MonitorInfo> resultDs = ds1.filter(new RichFilterFunction<MonitorInfo>() {

            ResultSet resultSet = null;
            PreparedStatement preparedStatement = null;
            Connection connection = null;


            @Override
            public void open(Configuration parameters) throws Exception {
                connection = JdbcUtils.getconnection();
                preparedStatement = connection.prepareStatement("select * from t_violation_list where car = ?");


            }

            @Override
            public void close() throws Exception {
                JdbcUtils.release(resultSet, preparedStatement, connection);
            }

            @Override
            public boolean filter(MonitorInfo value) throws Exception {
                preparedStatement.setString(1, value.getCar());
                resultSet = preparedStatement.executeQuery();
                return resultSet.next();
            }
        });

        resultDs.print();
        // 将数据放入hbase中

        // 可以通过遍历resultDs，然后通过hbase原生API将数据保存到hbase中。


        env.execute();

    }
}
